package com.peaksport.pkms.learntest.controller;

import javax.ws.rs.core.MediaType;

import org.apache.servicecomb.provider.rest.common.RestSchema;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestParam;

import com.peaksport.framework.base.controller.BaseController;
import com.peaksport.framework.extend.kafka.PKKafkaProducer;
import com.peaksport.pkms.oauth.vo.TokenVO;

@RestSchema(schemaId="pkms-test-sendKafka")
@RequestMapping(value="test/testKafka",produces=MediaType.APPLICATION_JSON)
public class TestKafaSendController extends BaseController {
	
	@Autowired
	private PKKafkaProducer kafkaSender;
	

	
	@GetMapping("sendJson")
	public String sendKafka(@RequestParam String topic,
							@RequestParam String message,
							@RequestParam(required=false) String key) {
		TokenVO vo = new TokenVO();
		vo.setClientId("abc");
		vo.setClientNum("guo111");
		kafkaSender.sendJsonMsg_sync(topic,null,key,message,null,null);
		return "发送json格式kafka消息";
	}
	
	@GetMapping("createTopic") 
	public String createTopic(@RequestParam String topic,@RequestParam int numPartitions,@RequestParam short replicationFactor) {
		kafkaSender.createTopic(topic, numPartitions, replicationFactor);
		return "创建kafka主题[" + topic + "]";
	}
	
	@GetMapping("existsTopic") 
	public String existsTopic(@RequestParam String topic) {
		boolean b = kafkaSender.existsTopic(topic);
		if (b) return "存在kafka主题[" + topic + "]";
		else return "不存在kafka主题[" + topic + "]";
	}
}
